﻿/*
 * Created by SharpDevelop.
 * User: Corrado Mio
 * Date: 02/06/2012
 * Time: 18:08
 * 
 * Implementazione degli oggetti fondamentali per l'utilizzo del protocollo PGM:
 * 
 *  PgmSocket           classe base
 *      PgmSender       invia i buffer usando il protocollo PGM
 *      PgmListener     si mette in ascolto in attesa di un fluss PGM
 *      PgmReceiver     riceve un flusso PGM
 * 
 * Dalle classi PgmListener e PgmReceiver e' stata implementata la classe
 * 
 *      PgmConnect
 * 
 * che ha il compito di ascoltare su un canale IP e ricevere il flusso dati, coordinando 
 * opportunamente un PgmListener e 0 o piu' PgmClient (un'estensione di PgmReceiver)
 * 
 * Nota: il protocollo PGM prevede tre entita':
 * 
 *  - un Sender
 *  - un Receiver
 *  - un Listener
 * 
 * 1) il sender invia i dati utilizzando il protocollo PGM
 * 2) il listener si mette in ascolto dei flusso PGM. Deve essere fatta un'accept per ogni
 *    sender, perche' ogni sender ha un proprio flusso di informazioni e correzioni
 * 3) il receiver riceve i dati sa un particolare sender. 
 * 
 */
using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.Linq.Expressions;
using System.Net;
using System.Net.Sockets;
using System.Runtime.InteropServices;
using System.Threading;

namespace CSharpExt.Net.Sockets
{
    // ----------------------------------------------------------------------
    // Enumerations
    //  ----------------------------------------------------------------------

    public enum FecMode
    {
        #region Constants

        Disabled, 
        ProActive, 
        OnDemand, 
        Both 

        #endregion
    }

    public enum DataMode
    {
        #region Constants

        Message = SocketType.Rdm, 
        Stream  = SocketType.Stream 

        #endregion
    }

    // ----------------------------------------------------------------------
    // Data Structures
    //  ----------------------------------------------------------------------

    enum RmOptionName
    {
        #region Constants
        
        RM_RATE_WINDOW_SIZE = 1001, 
        RM_SET_MESSAGE_BOUNDARY, 
        RM_FLUSHCACHE, 
        RM_SENDER_WINDOW_ADVANCE_METHOD, 
        RM_SENDER_STATISTICS, 
        RM_LATEJOIN, 
        RM_SET_SEND_IF, 
        RM_ADD_RECEIVE_IF, 
        RM_DEL_RECEIVE_IF, 
        RM_SEND_WINDOW_ADV_RATE, 
        RM_USE_FEC, 
        RM_SET_MCAST_TTL, 
        RM_RECEIVER_STATISTICS, 
        RM_HIGH_SPEED_INTRANET_OPT, 

        #endregion
    }

    /// <summary>
    /// WindowSizeInBytes = (RateKbitsPerSec / 8) * WindowSizeInMSecs
    /// </summary>
    struct _RM_SEND_WINDOW
    {
        #region Data Structure

        public ulong RateKbitsPerSec; // Send rate
        public ulong WindowSizeInMSecs;
        public ulong WindowSizeInBytes;

        #endregion
    }

    struct _RM_FEC_INFO
    {
        #region Data Structure

        public ushort FECBlockSize;
        public ushort FECProActivePackets;
        public byte   FECGroupSize;
        public byte   FECOnDemandParityEnabled;

        #endregion
    }

    // ----------------------------------------------------------------------
    // PgmSocket 
    //  ----------------------------------------------------------------------

    public abstract class PgmSocket : IDisposable
    {
        #region Constants

        private const int IPPROTO_RM = 113;

        public const ProtocolType      PgmProtocolType = (ProtocolType)IPPROTO_RM;
        public const SocketOptionLevel PgmSocketOption = (SocketOptionLevel)IPPROTO_RM;

        public const int SENDER_DEFAULT_RATE_KBITS_PER_SEC = 56;
        public const int SENDER_DEFAULT_WINDOW_SIZE_BYTES  = 10 * 1000 * 1000;

        public const int SENDER_DEFAULT_WINDOW_ADV_PERCENTAGE = 15;
        public const int MAX_WINDOW_INCREMENT_PERCENTAGE      = 25;

        public const int MAX_MCAST_TTL = 1;

        public const int SENDER_DEFAULT_LATE_JOINER_PERCENTAGE = 0;
        public const int SENDER_MAX_LATE_JOINER_PERCENTAGE     = 75;

        public const int FEC_GROUP_SIZE = 64;
        public const int FEC_BLOCK_SIZE = 128;
        public const int FEC_PROACTIVE_PACKETS  = 128;

        #endregion

        #region Constants/2

        public const long    DefaultSendRate            = 5 * 1024; // 5Mbps
        public const int     DefaultSendTimeout         = 1000;     // 1s
        public const long    DefaultBufferSize          = 16 * 1024 * 1024; // 10 MB
        public const FecMode DefaultFecMode             = FecMode.Disabled;
        public const int     DefaultLateJoinPercentage  = SENDER_DEFAULT_LATE_JOINER_PERCENTAGE;
        public const int     DefaultMcastTtl            = MAX_MCAST_TTL;
        public const int     DefaultWindowAdvPercentage = SENDER_DEFAULT_WINDOW_ADV_PERCENTAGE;

      //public const long    MaxBufferPoolSize          = 64 * 1024;
      //public const int     MaxReceivedMessageSize     = 64 * 1024;
      //public const int     LateJoin                   = 50;
      //public const bool    SetMmgBoundary             = false;
      //public const DataMode DefaultDataMode           = DataMode.Message;

        #endregion
        
        #region Protected Data

        private const int SO_MAX_MSG_SIZE = 0x2003;
        
        protected DataMode   m_mode = DataMode.Message;
        
        protected Socket     _socket;
        
        protected Disposed _disposed = new Disposed();
        
        #endregion

        #region Constructor

        protected PgmSocket()
        {
            
        }

//        protected PgmSocket(String ip, int port, DataMode mode) : this(new IPEndPoint(IPAddress.Parse(ip), port), mode)
//        {
//            
//        }
        
//        protected PgmSocket(IPEndPoint ipep, DataMode mode)
//        {
//            m_mode = mode;
//            
//            EndPoint = ipep;
//            
//            CreateSocket();
//        }
        
        // ------------------------------------------------------------------

        protected Socket CreateSocket(SocketType mode)
        {
            if (mode != SocketType.Rdm && mode != SocketType.Stream)
            {
                throw new ArgumentException("PGM socket type must be set to SocketType.Rdm or SocketType.Stream", "mode");
            }

            return new Socket(AddressFamily.InterNetwork, mode, PgmProtocolType);
        }

        protected void CreateSocket()
        {
            _socket = CreateSocket((SocketType)m_mode);
            
            _socket.SetSocketOption(SocketOptionLevel.Socket, SocketOptionName.ReuseAddress, true);
        }
        
        #endregion
        
        #region Properties

        public Socket Socket
        {
            get { return _socket; }
        }

        
        public IPAddress  NetworkInterface { get; set; }
        
        public IPEndPoint EndPoint { get; set; }
        
        
        public bool Opened
        {
            get { return _socket != null && (_socket.Connected || _socket.IsBound); }
        }
        
        public int ProtocolMaxMessageSize
        {
            get { return (int) _socket.GetSocketOption(SocketOptionLevel.Socket, (SocketOptionName) SO_MAX_MSG_SIZE); }
        }

        #endregion

        #region Operations

        public abstract void Open();
        
        protected abstract void Configure();

        public void Close()
        {
            Dispose(true);
        }
        
        protected virtual void Dispose(bool disposing)
        {
            if(_disposed.CanDispose)
                OnClose();
        }
        
        protected virtual void OnClose()
        {
            try { _socket.Shutdown(SocketShutdown.Both); } catch(Exception) { }
            _socket.Close();
            _socket = null;
        }
        
        #endregion

        #region IDisposable

        public void Dispose()
        {
            Dispose(true);
        } 

        #endregion
        
        #region Protected Methods
        
        
        #endregion
        
        #region Implementation
        
        internal static byte[] StructureToByteArray(Object obj)
        {
            int len = Marshal.SizeOf(obj);
            byte[] result = new byte[len];
            IntPtr ptr = Marshal.AllocHGlobal(len);
            Marshal.StructureToPtr(obj, ptr, true);
            Marshal.Copy(ptr, result, 0, len);
            Marshal.FreeHGlobal(ptr);

            return result;
        }

        internal static T ByteArrayToStructure<T>(byte[] array)
        {
            T result = default(T);
            int len = Marshal.SizeOf(typeof(T));
            IntPtr ptr = Marshal.AllocHGlobal(len);
            Marshal.Copy(array, 0, ptr, len);
            result = (T)Marshal.PtrToStructure(ptr, typeof(T));
            Marshal.FreeHGlobal(ptr);

            return result;
        }

        #endregion
    }
    
    // ----------------------------------------------------------------------
    // PgmSender
    //  ----------------------------------------------------------------------

    public class PgmSender : PgmSocket
    {
        #region Constructor

        public PgmSender() : base()
        {
            SetDefaults();
        }
        
        //  -----------------------------------------------------------------

        protected virtual void SetDefaults()
        {
            FecMode         = DefaultFecMode;
            SendRateKbps    = DefaultSendRate;
            SendBufferSize  = 0;//DefaultBufferSize;

            Ttl             = DefaultMcastTtl;
            SendTimeout     = DefaultSendTimeout;
            
            LateJoinPercentage        = DefaultLateJoinPercentage;
            WindowIncrementPercentage = DefaultWindowAdvPercentage;
            
            MaxAsync        = 1;
        }

        #endregion

        #region Properties

        public FecMode FecMode { get; set; }
        public ulong   SendRateKbps { get; set; }
        public ulong   SendBufferSize { get; set; }
        public bool    SetMessageBoundary { get; set; }
        public int     LateJoinPercentage { get; set; }
        public int     WindowIncrementPercentage { get; set; }
        
        public int     Ttl { get; set; }
        public int     SendTimeout { get; set; }
        
        public int     MaxAsync { get; set; }

        public SocketFlags SocketFlags { get; set; }
        
        #endregion

        #region Operations
        
        public override void Open()
        {
            if (_socket == null)
                CreateSocket();

            Bind();
            Configure();
            Connect();
        }

        protected override void Configure()
        {
            SetSendWindow();
            SetFec();
            SetLateJoin();
            SetWindowIncrement();
            SetMulticastTTL();
            SetSenderInterface();
        }

        #endregion

        #region Send

        public int Send(byte[] data, int offset, int count)
        {
            BeforeSend(count - offset);
            return _socket.Send(data, offset, count, SocketFlags);
        }

        public IAsyncResult BeginSend(byte[] data, int offset, int count, AsyncCallback callback, Object state)
        {
            BeforeSend(count - offset);
            return _socket.BeginSend(data, offset, count, SocketFlags, callback, state);
        }

        public int EndSend(IAsyncResult result, out SocketError errCode)
        {
            return _socket.EndSend(result, out errCode);
        }

        #endregion
        
        #region Implementation
        
        private void Bind()
        {
            IPEndPoint bind = new IPEndPoint(IPAddress.Any, 0);
            
            // non sembra sia possibile fare la bind su un particolare IP
            // e non ha nessun effetto indicare esplicitamene la porta: mostra 
            // sempre 0.
            //
            // La Bind serve? La Bind SERVE! ed a quando sembra, DEVE essere
            // fatta su (0.0.0.0:0)
            //
            //if (NetworkInterface == null)
            //    bind = new IPEndPoint(IPAddress.Any, EndPoint.Port);
            //else
            //    bind = new IPEndPoint(NetworkInterface, 0);
            
            _socket.Bind(bind);
        }
        
        private void Connect()
        {
            _socket.Connect(EndPoint);
        }
        

        private void SetSenderInterface()
        {
            if (NetworkInterface != null)
            {
                byte[] ipBytes = NetworkInterface.GetAddressBytes();
                _socket.SetSocketOption(PgmSocketOption, (SocketOptionName)RmOptionName.RM_SET_SEND_IF, ipBytes);
            }
        }

        private void SetSendWindow()
        {
            if (SendBufferSize == 0)
                SendBufferSize = SendRateKbps*1024u/8u*(uint)SendTimeout/1000u;
            
            _RM_SEND_WINDOW sndWnd = new _RM_SEND_WINDOW();
            sndWnd.WindowSizeInMSecs = SendBufferSize; //SENDER_DEFAULT_WINDOW_SIZE_BYTES;
            //sndWnd.WindowSizeInMSecs = (ulong)SendTimeout;
            sndWnd.RateKbitsPerSec   = SendRateKbps;

            byte[] structBuf = StructureToByteArray(sndWnd);
            _socket.SetSocketOption(PgmSocketOption, (SocketOptionName)RmOptionName.RM_RATE_WINDOW_SIZE, structBuf);
        }

        private void SetLateJoin()
        {
            _socket.SetSocketOption(PgmSocketOption, (SocketOptionName)RmOptionName.RM_LATEJOIN, LateJoinPercentage);
        }

        private void SetFec()
        {
            if (FecMode == FecMode.Disabled)
                return;
            
            _RM_FEC_INFO fec = new _RM_FEC_INFO();
            fec.FECGroupSize = FEC_GROUP_SIZE;
            fec.FECBlockSize = FEC_BLOCK_SIZE;

            switch (FecMode)
            {
                //case FecMode.Disabled:
                //    return;
                case FecMode.ProActive:
                    fec.FECProActivePackets = FEC_PROACTIVE_PACKETS;
                    break;
                case FecMode.OnDemand:
                    fec.FECOnDemandParityEnabled = 1;
                    break;
                case FecMode.Both:
                    fec.FECProActivePackets = FEC_PROACTIVE_PACKETS;
                    fec.FECOnDemandParityEnabled = 1;
                    break;
                default:
                    return;
            }

            byte[] structBuf = StructureToByteArray(fec);
            _socket.SetSocketOption(PgmSocketOption, (SocketOptionName)RmOptionName.RM_USE_FEC, structBuf);
        }

        // --------------------------------------------------------------------

        private void BeforeSend(int msgSize)
        {
            if (SetMessageBoundary)
                SetMsgBoundary(msgSize);

            _socket.SendTimeout = SendTimeout;
        }

        private void SetMsgBoundary(int boundary)
        {
            _socket.SetSocketOption(PgmSocketOption, (SocketOptionName)RmOptionName.RM_SET_MESSAGE_BOUNDARY, boundary);
        }

        private void SetWindowIncrement()
        {
            _socket.SetSocketOption(PgmSocketOption, (SocketOptionName)RmOptionName.RM_SEND_WINDOW_ADV_RATE, WindowIncrementPercentage);
        }

        private void SetMulticastTTL()
        {
            _socket.SetSocketOption(PgmSocketOption, (SocketOptionName)RmOptionName.RM_SET_MCAST_TTL, Ttl);
        }

        #endregion
    }

    // ----------------------------------------------------------------------
    // PgmListener
    // PgmReceiver
    //  ----------------------------------------------------------------------

    public class PgmListener : PgmSocket
    {
        #region Constructor

        public PgmListener() : base()
        {

        }
        
        #endregion

        #region Operations
        
        public override void Open()
        {
            if (_socket == null)
                CreateSocket();

            _socket.Bind(EndPoint);
            
            Configure();
            
            _socket.Listen(16/*backlog*/);
        }

        protected override void Configure()
        {
            SetHighSpeedNetwork();
            SetReceiverInterface();
        }

        #endregion

        #region Accept

        public Socket AcceptSocket()
        {
            return _socket.Accept();
        }

        public PgmReceiver Accept()
        {
            Socket receiver = _socket.Accept();

            return CreateReceiver(receiver);
        }

        // --------------------------------------------------------------------

        public IAsyncResult BeginAccept(AsyncCallback callback, Object state)
        {
            return _socket.BeginAccept(callback, state);
        }

        public PgmReceiver EndAccept(IAsyncResult result)
        {
            Socket receiver = _socket.EndAccept(result);
            
            return CreateReceiver(receiver);
        }

        public Socket EndAcceptSocket(IAsyncResult result)
        {
            return _socket.EndAccept(result);
        }

        #endregion
        
        #region Implementation

        private PgmReceiver CreateReceiver(Socket socket)
        {
            PgmReceiver receiver = new PgmReceiver(socket);
            receiver.Open();
            return receiver;
        }
        
        protected void SetHighSpeedNetwork()
        {
            try
            {
                Socket.SetSocketOption(PgmSocketOption, (SocketOptionName)RmOptionName.RM_HIGH_SPEED_INTRANET_OPT, 1);
            }
            catch (SocketException ex)
            {
                Debug.WriteLine("Failed to set highspeed network : " + ex.Message);
            }
        }

        private void SetReceiverInterface()
        {
            if (NetworkInterface != null)
                AddReceiverInterface(NetworkInterface);
        }

        // --------------------------------------------------------------------

        private void AddReceiverInterface(IPAddress netitfc)
        {
            byte[] ipbytes = netitfc.GetAddressBytes();
            _socket.SetSocketOption(PgmSocketOption, (SocketOptionName)RmOptionName.RM_ADD_RECEIVE_IF, ipbytes);
        }

        private void DeleteReceiverInterface(IPAddress netitfc)
        {
            byte[] ipbytes = netitfc.GetAddressBytes();
            _socket.SetSocketOption(PgmSocketOption, (SocketOptionName)RmOptionName.RM_DEL_RECEIVE_IF, ipbytes);
        }
        
        #endregion
    }
    
    public class PgmReceiver : PgmSocket
    {
        #region Private Data
        
        #endregion

        #region Constructor

        public PgmReceiver(Socket socket)
        {
            _socket = socket;
        }

        #endregion

        #region Properties

        public EndPoint RemoteEndPoint
        {
            get { return Socket.RemoteEndPoint; }
        }

        #endregion

        #region Operations
        
        public override void Open()
        {
            Configure();
        }
        
        protected override void Configure()
        {
            SetHighSpeedNetwork();
        }

        #endregion

        #region Receive

        public int Receive(byte[] data, int offset, int count)
        {
            return _socket.Receive(data, offset, count, SocketFlags.None);
        }

        public IAsyncResult BeginReceive(byte[] data, int offset, int count, AsyncCallback callback, Object state)
        {
            return _socket.BeginReceive(data, offset, count, SocketFlags.None, callback, state);
        }
        
        public int EndReceive(IAsyncResult result, out SocketError errorCode)
        {
            return _socket.EndReceive(result, out errorCode);
        }
        
        
        public int ReceiveFrom(byte[] data, int offset, int count, ref EndPoint fromep)
        {
            fromep = RemoteEndPoint;
            
            return _socket.Receive(data, offset, count, SocketFlags.None);
        }

        #endregion
        
        #region Implementation
        
        protected void SetHighSpeedNetwork()
        {
            try
            {
                Socket.SetSocketOption(PgmSocketOption, (SocketOptionName)RmOptionName.RM_HIGH_SPEED_INTRANET_OPT, 1);
            }
            catch (SocketException ex)
            {
                Debug.WriteLine("Failed to set highspeed network : " + ex.Message);
            }
        }

        #endregion
    }

    // ----------------------------------------------------------------------
    // PgmConnect
    //  ----------------------------------------------------------------------
    // La classe coordina un PgmListener ed una lista di PgmClient.
    //
    // Nota: il listenere ha il compito di creare un socket per ogni sender
    // e questo e' ovvio perche' c'e' una comunicazione dedicata con correzione
    // di errore per ogni sender.
    // Inoltre, la perdita di pachetti potrebbe avvenire solo su un socket ma
    // non sugli altri
    //
    // Nota: come implementare una Receive sincrona?
    //
    //  Receive
    //  BeginReceive
    //  EndReceive
    
    public class PgmConnect
    {
        #region Private Data
        
        private IPEndPoint _endPoint;
        
        private IPAddress _interface;
        
        // ------------------------------------------------------------------
        
        private Thread _listenThread;
        
        private PgmListener _listener;
        
        private List<PgmClient> _clients = new List<PgmClient>();
        
        private Disposed _disposed = new Disposed();
        
        #endregion
        
        #region Constructor
        
        public PgmConnect()
        {
            
        }
        
        #endregion
        
        #region Properties
        
        public Socket Socket
        {
            get { return _listener.Socket; }
        }
        
        public IPEndPoint EndPoint
        {
            get { return _endPoint;  }
            set { _endPoint = value; }
        }
        
        public IPAddress NetworkInterface
        {
            set { _interface = value; }
        }
        
        // ------------------------------------------------------------------
        
        public bool Opened
        {
            get { return _listener != null && _listener.Opened; }
        }
        
        // ------------------------------------------------------------------
        
        public Action<byte[], int, int> OnReceived;
        
        public Action<PgmClient, byte[], int, int, EndPoint> OnReceivedFrom;
        
        public Action<PgmClient, Exception> OnClientException;

        // ------------------------------------------------------------------
        
        public Action<PgmClient, bool> OnClient;
        
        public Action<Exception> OnException;
        
        #endregion
        
        #region Operations
        
        public void Open()
        {
            OnOpen();
        }
        
        public void Close()
        {
            if(_disposed.CanDispose)
                OnClose();
        }
        
        protected virtual void OnOpen()
        {
            createListener();
            startListener();
        }
        
        protected virtual void OnClose()
        {
            closeListener();
            stopListener();
        }
        
        #endregion
        
        #region Implementation
        
        protected Thread CreateThread(ThreadStart listen)
        {
            return new Thread(listen);
        }
        
        protected PgmClient CreateClient(Socket receiver)
        {
            return new PgmClient(receiver);
        }
        
        // ------------------------------------------------------------------
        
        private void createListener()
        {
            _listener = new PgmListener()
            {
                EndPoint = _endPoint, 
                NetworkInterface = _interface, 
            };
            
            _listener.Open();
        }
        
        private void closeListener()
        {
            _listener.Close();
        }
        
        private void startListener()
        {
            _listenThread = CreateThread(listen);
            _listenThread.Start();
        }
        
        private void stopListener()
        {
            if(_listenThread != null && !_listenThread.Join(1000))
                _listenThread.Abort();
            _listenThread = null;
        }
        
        private void listen()
        {
            try
            {
                while(Opened)
                {
                    Socket socket = null;
                    PgmClient client = null;
                    
                    try
                    {
                        IAsyncResult ar = _listener.BeginAccept(null, null);
                        
                        while(Opened && !ar.AsyncWaitHandle.WaitOne(500))
                        {
                            checkClients();
                        }
                        
                        if (!Opened) 
                            continue;
                        
                        socket = _listener.EndAcceptSocket(ar);
                    }
                    catch (Exception e)
                    {
                        if (OnException != null)
                            OnException(e);
                        
                        break;
                    }
                    
                    try
                    {
                        client = CreateClient(socket);
                        
                        initClient(client);
                        
                        client.Open();
                        
                        addClient(client);
                    }
                    catch (Exception e)
                    {
                        if (OnClientException != null)
                            OnClientException(client, e);
                    }
                }
            }
            finally
            {
                //_listenThread = null;
                
                Close();
            }
            
        }
        
        // ------------------------------------------------------------------
        
        private void initClient(PgmClient receiver)
        {
            receiver.EndPoint = EndPoint;
            receiver.OnReceived = OnReceived;
            receiver.OnReceivedFrom = OnReceivedFrom;
            receiver.OnException = OnClientException;
        }
        
        private void addClient(PgmClient receiver)
        {
            lock (_clients)
            {
                _clients = new List<PgmClient>(_clients);
                _clients.Add(receiver);
            }
            
            if (OnClient != null)
                OnClient(receiver, true);
        }
        
        private void removeClient(PgmClient receiver)
        {
            if (OnClient != null)
                OnClient(receiver, false);
            
            lock (_clients)
            {
                _clients = new List<PgmClient>(_clients);
                _clients.Remove(receiver);
            }
        }
        
        private void checkClients()
        {
            foreach(PgmClient receiver in _clients)
            {
                if (!receiver.Opened)
                    removeClient(receiver);
            }
        }
        
        #endregion
    }
    
    
    
    public class PgmClient : PgmReceiver//: IDiagnose
    {
        #region Private Data
        
        private Thread _receiveThread;
        
        private Exception _exception;
        
        #endregion

        #region Constructor

        public PgmClient(Socket socket) : base(socket)
        {
            
        }

        #endregion

        #region Properties
        
        public Exception CatchedException
        {
            get { return _exception; }
        }

        // ------------------------------------------------------------------
        // E' possibile attivare una ricezione asincrona
        // Se la ricezione fallisce, il thread viene terminato ed il socket 
        // viene chiuso (viene chiamata la 'Disconnect()' ).
        // 
        
        /// <summary>
        /// Callback chiamate dai client alla ricezione di un buffer
        /// </summary>
        public Action<byte[], int, int> OnReceived;
        
        public Action<PgmClient, byte[], int, int, EndPoint> OnReceivedFrom;
        
        public Action<PgmClient, Exception> OnException;

        #endregion

        #region Operations
        
        public override void Open()
        {
            base.Open();
            
            if (OnReceived != null)
                OnReceivedFrom = onReceivedFrom;
            
            if (OnReceivedFrom != null)
                Start();
        }
        

        #endregion

        #region Implementation
        
        protected Thread CreateThread(ThreadStart start)
        {
            return new Thread(start);
        }

        // ------------------------------------------------------------------
        
        private void Start()
        {
            _receiveThread = CreateThread(receive);
            _receiveThread.Start();
        }
        
        private void Stop()
        {
            if (_receiveThread != null && !_receiveThread.Join(1000))
                _receiveThread.Abort();
            _receiveThread = null;
        }
        
        private void receive()
        {
            byte[] buffer = new byte[65536];
            
            EndPoint fromep = new IPEndPoint(0,0);
                    
            try
            {
                while(Opened)
                {
                    try
                    {
                        int readed = ReceiveFrom(buffer, 0, 65536, ref fromep);
                        
                        OnReceivedFrom(this, buffer, 0, readed, fromep);
                    }
                    catch (Exception e)
                    {
                        onException(e);
                        
                        break;
                    }
                }
            }
            finally
            {
                Close();
            }
        }
        
        private void onReceivedFrom(PgmClient receiver, byte[] buffer, int offset, int count, EndPoint fromep)
        {
            OnReceived(buffer, offset, count);
        }
        
        private void onException(Exception e)
        {
            _exception = e;
                        
            if (OnException != null)
                OnException(this, e);
        }
        
        #endregion
    }
    
}
